package com.link.serve.config;
import com.link.serve.kafka.proto.KafkaMessageProto.KafkaMessage;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import java.util.HashMap;
import java.util.Map;
@Configuration
public class KafkaConfig {
    @Value("${redis.kafka.consumer.key-deserializer}")// yml中读取消费者反序列化器的全限定类名
    private String consumerKeyDeserializerClassName;
    @Value("${redis.kafka.consumer.value-deserializer}")
    private String consumerValueDeserializerClassName;
    @Value("${spring.kafka.bootstrap-servers}")// 全局 Kafka 集群地址
    private String bootstrapServers;
    @Value("${spring.kafka.consumer.group-id}")// 全局 Kafka 消费者组ID
    private String groupId;

    @Bean//动态创建 ConsumerFactory，利用 yml 中自定义的反序列化器配置。
    public ConsumerFactory<String, KafkaMessage> consumerFactory() throws ClassNotFoundException {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); // kafka地址
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);                   // 消费者组id
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);           // 是否启动自动提交偏移量
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);                // 最大批量拉取消息数
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, Class.forName(consumerKeyDeserializerClassName));    //配置key的反序列化器
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, Class.forName(consumerValueDeserializerClassName));//配置value的反序列化器

        return new DefaultKafkaConsumerFactory<>(props);// 返回生成的 ConsumerFactory 配置工厂
    }

    /**
     * 创建批量消费的 Kafka Listener 容器工厂
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, KafkaMessage> kafkaBatchListenerContainerFactory(ConsumerFactory<String, KafkaMessage> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, KafkaMessage> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setBatchListener(true);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        // 设置 idleBetweenPolls 等其他参数，如需要
        factory.getContainerProperties().setIdleBetweenPolls(10_000); // 例如 10 秒
        return factory;
    }
}
